package cn.selfiot.iotman.mqtt;

import cn.selfiot.iotman.helper.SSLHelper;
import cn.selfiot.iotman.model.MqttProjectConstants;
import cn.selfiot.iotman.model.mqtt.*;
import cn.selfiot.iotman.model.mqtt.messge.PublishMessageModel;
import cn.selfiot.iotman.model.mqtt.messge.SubscribeTopicModel;
import cn.selfiot.iotman.model.mqtt.ssl.CACertFileModel;
import cn.selfiot.iotman.model.mqtt.ssl.SelfSignedCertModel;
import org.bouncycastle.jce.provider.BouncyCastleProvider;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.net.SocketFactory;
import javax.net.ssl.SSLSocketFactory;
import java.nio.charset.StandardCharsets;
import java.security.Security;

/**
 * @author hugo
 * @since 2020/8/13
 */
public class MqttClientManager {
    private static final Logger LOGGER = LoggerFactory.getLogger(MqttClientManager.class);
    
    //    private static MqttClientManager INSTANCE;
    private static IMqttClient mqttClient;
    private MqttConnectOptions connOpts;
    
    private MqttClientManager(ProjectModel project) {
        if (project != null) {
            boolean enableSSL = project.getSsl() != null
                    && project.getSsl().getEnable();
            String protocol = enableSSL ? "ssl://" : "tcp://";
            String hostUrl = protocol + project.getHostAddr() + ":"
                    + project.getHostPort();
            try {
                mqttClient = new MqttClient(hostUrl, project.getClientId(), new MqttDefaultFilePersistence());
            } catch (MqttException e) {
                LOGGER.error(e.getMessage());
            }
            connOpts = initConnOption(project);
        } else {
            LOGGER.error("project info is null!");
        }
    }
    
    public static MqttClientManager getInstance(ProjectModel project) {
//        if (null == INSTANCE) {
//            synchronized (MqttClientManager.class) {
//                INSTANCE = new MqttClientManager(project);
//            }
//        }
//        return INSTANCE;
        return new MqttClientManager(project);
    }
    
    /**
     * 连接选项设置
     *
     * @return 连接选项值对象
     */
    private MqttConnectOptions initConnOption(ProjectModel project) {
        MqttConnectOptions connOptions = new MqttConnectOptions();
        ConnParamsModel connParams = project.getConnParams();
        UserPwdAuthModel authModel = project.getUserPwdAuth();
        
        Boolean cleanSession = connParams.getCleanSession();
        Boolean reconn = connParams.getReconnection();
        connOptions.setCleanSession(cleanSession == null ? false : cleanSession);
        connOptions.setAutomaticReconnect(reconn == null ? false : reconn);
        connOptions.setConnectionTimeout(connParams.getConnectionTimeout());
        connOptions.setKeepAliveInterval(connParams.getKeepAlive());
        connOptions.setMaxInflight(connParams.getMaxInflight());
        if (authModel != null) {
            connOptions.setUserName(authModel.getUserName());
            connOptions.setPassword(authModel.getPassword().toCharArray());
        }
        
        int mqttVer = 0;
        if (connParams.getMqttVersion()
                .equals(MqttProjectConstants.MQTT_VER[0].getValue())) {
            mqttVer = MqttConnectOptions.MQTT_VERSION_3_1;
        }
        if (connParams.getMqttVersion()
                .equals(MqttProjectConstants.MQTT_VER[1].getValue())) {
            mqttVer = MqttConnectOptions.MQTT_VERSION_3_1_1;
        }
        connOptions.setMqttVersion(mqttVer);
        if (project.getSsl().getEnable()) {
            SocketFactory socketFactory = this.getSocketFactory(project);
            if (socketFactory != null)
                connOptions.setSocketFactory(socketFactory);
        }
        setWill(connOptions, project);
        
        return connOptions;
    }
    
    // 设置Will消息，意外掉线后通知平台
    private void setWill(MqttConnectOptions connOptions, ProjectModel project) {
        LwtModel lwt = project.getLwt();
        if (lwt != null) {
            try {
                connOptions.setWill(lwt.getTopic(),
                        lwt.getPayload().getBytes(),
                        lwt.getQos(),
                        lwt.getRetained());
            } catch (Exception e) {
                LOGGER.error("Exception Occurred {}", e.getMessage());
            }
        } else {
            LOGGER.info("lwt not set.");
        }
    }
    
    private SocketFactory getSocketFactory(ProjectModel project) {
        Security.addProvider(new BouncyCastleProvider());
        SSLSocketFactory socketFactory = null;
        SSLModel sslModel = project.getSsl();
        if (sslModel.getCaCertFile() != null) {
            CACertFileModel caCertFile = sslModel.getCaCertFile();
            try {
                socketFactory = SSLHelper.getSSLSocketFactory(caCertFile.getFilePath());
            } catch (Exception e) {
                LOGGER.error(e.getMessage());
            }
        }
        if (sslModel.getCaKeystore() != null) {
            try {
                socketFactory = SSLHelper.getSSLSocketFactory(
                        sslModel.getCaKeystore().getKeystorePwd(),
                        sslModel.getCaKeystore().getFilePath());
            } catch (Exception e) {
                LOGGER.error(e.getMessage());
            }
        }
        if (sslModel.getSelfSignedCert() != null) {
            SelfSignedCertModel selfSigned = sslModel.getSelfSignedCert();
            try {
                socketFactory = SSLHelper.getSSLSocketFactory(
                        selfSigned.getCaFile(),
                        selfSigned.getClientCertFile(),
                        selfSigned.getClientKeyFile(),
                        selfSigned.getClientKeyPwd());
            } catch (Exception e) {
                LOGGER.error(e.getMessage());
            }
        }
        
        return socketFactory;
    }
    
    /**
     * 连接MQTT Broker
     *
     * @throws MqttException 异常
     */
    public void connect(MqttCallbackExtended mqttCallback) throws MqttException {
        if (!mqttClient.isConnected()) {
            mqttClient.setCallback(mqttCallback);
            IMqttToken token = mqttClient.connectWithResult(connOpts);
            token.waitForCompletion();
        }
    }
    
    /**
     * 断开连接
     *
     * @throws MqttException 异常
     */
    public static void disconnect() throws MqttException {
        mqttClient.disconnect();
    }
    
    public boolean isConnected() {
        return mqttClient.isConnected();
    }
    
    /**
     * 发布消息
     *
     * @param publishMessage 发布消息对象
     * @throws MqttException 异常
     */
    public void publishMessage(PublishMessageModel publishMessage)
            throws MqttException {
        mqttClient.publish(
                publishMessage.getTopic().getTopicString(),
                publishMessage.getPayload().getBytes(StandardCharsets.UTF_8),
                publishMessage.getQos(),
                publishMessage.getRetained());
    }
    
    /**
     * 订阅主题，
     *
     * @throws MqttException 抛出异常
     */
    public void subscribe(SubscribeTopicModel subscribeTopic,IMqttActionListener listener)
            throws MqttException {
        IMqttToken token = mqttClient.subscribeWithResponse(
                subscribeTopic.getTopic().getTopicString(),
                subscribeTopic.getQos());
        token.setActionCallback(listener);
    }
    
    /**
     * 取消订阅
     *
     * @throws MqttException 抛出异常
     */
    public void unSubscribe(final String topic) throws MqttException {
        mqttClient.unsubscribe(topic);
    }
    
    
}
